package com.alibaba.otter.node.etl.load.loader.mq.rabbit.api;

import com.alibaba.otter.node.etl.load.loader.mq.rabbit.MqConstants;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.MqUtils;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.callback.ConnectedCallback;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.callback.DisconnectedCallback;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.model.RabbitMqNamespace;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.namespace.NamespaceManager;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.resolver.RabbitMqNamespaceResolver;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ShutdownListener;
import com.rabbitmq.client.ShutdownSignalException;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class ChannelPool {

    @Getter
    private int minSize;

    @Getter
    private int coreSize;

    @Getter
    private int maxSize;

    @Setter
    private ConnectedCallback connectedCallback;

    @Setter
    private DisconnectedCallback disconnectedCallback;

    @Setter
    private long retryInterval = MqConstants.RECREATE_CHANNEL_INTERVAL_IN_MS;

    private AtomicInteger size = new AtomicInteger(0);

    @Getter
    private ConcurrentHashMap<Integer, Channel> allChannels = new ConcurrentHashMap<>();
    private ConcurrentHashMap<Integer, Channel> busyChannels = new ConcurrentHashMap<>();

    private BlockingQueue<Channel> idleChannels;
    private static RabbitMqConnectionManager connectionManager = new RabbitMqConnectionManager();
    private String namespace;

    public ChannelPool(String namespaceStr) {

        this.namespace = namespaceStr;

        RabbitMqNamespaceResolver resolver = new RabbitMqNamespaceResolver(namespaceStr);
        RabbitMqNamespace namespace = (RabbitMqNamespace) NamespaceManager.get(resolver);

        this.minSize = namespace.getSetting().getMinChannelSize();
        this.coreSize = namespace.getSetting().getCoreChannelSize();
        this.maxSize = namespace.getSetting().getMaxChannelSize();

        this.idleChannels = new ArrayBlockingQueue<>(this.maxSize);
    }

    public Channel getChannel() {
        try {

            if (size.get() < minSize) {
                // Channel数量小于MinSize时，增加到最小Size。
                initMinChannels();
            }

            if (size.get() < coreSize) {
                // Channel数量小于CoreSize时，直接增加。
                createChannel();
            }
            Channel channel;
            if (size.get() < maxSize) {
                // Channel数量小于MaxSize时，若能取到空闲Channel则直接返回，否则创建后，再重新获取。
                channel = idleChannels.poll();
                while (channel == null) {
                    createChannel();
                    channel = idleChannels.poll();
                }
            } else {
                // Channel数量大于MaxSize时，阻塞等待空闲Channel。
                long t1 = System.currentTimeMillis();
                channel = idleChannels.take();
                long t2 = System.currentTimeMillis();
                long waitTime = t2 - t1;
                if (waitTime > 10) {
                    // 等待超过阈值，则发出警告。
                    log.warn("Channel池中的channel达到最大数量[{}]，等待空闲Channel花费{}ms。"
                        + "尝试调节[mq.max-channel-size]配置来增加channel数量。", maxSize, waitTime);
                }
            }

            // 检查Channel是否可用，若不可用，则重新创建。
            if (channel.isOpen()) {
                busyChannels.put(channel.getChannelNumber(), channel);
            } else {
                // 如果channel不可用，则清理整个queue。
                log.info("Reset channel pool");
                reset();
                channel = getChannel();
            }
            return channel;

        } catch (InterruptedException e) {

            log.warn("等待空闲Channel出错，重新尝试。" + e.getMessage());
            return getChannel();
        }
    }

    @AllArgsConstructor
    class MyShutdownListener implements ShutdownListener {

        private Channel channel;

        @Override
        public void shutdownCompleted(ShutdownSignalException cause) {
            disconnectedCallback.setChannel(channel);
            disconnectedCallback.onDisconnected(cause);
            reset();
        }
    }

    private synchronized void initMinChannels() {
        if (size.get() >= minSize) {
            return;
        }
        for (int i = 0; i < minSize; i++) {
            boolean created = createChannel();
            if (!created) {
                log.warn("minSize大于maxSize，请检查配置。");
                break;
            }
        }
    }

    /**
     * 创建Channel，只有创建成功，且最终加入到缓存中才会返回true。
     */
    private synchronized boolean createChannel() {
        if (size.get() >= maxSize) {
            log.debug("Channel池中的channel达到最大数量[{}]。", maxSize);
            return false;
        }
        try {
            Channel channel = connectionManager.getConnection(namespace).createChannel();
            connectedCallback.connectedCallback(channel);
            channel.addShutdownListener(new MyShutdownListener(channel));

            size.incrementAndGet();
            boolean notFull = idleChannels.offer(channel);

            if (notFull) {
                allChannels.put(channel.getChannelNumber(), channel);
                return true;
            } else {
                // 理论上不会触发。
                closeChannel(channel);
                return false;
            }
        } catch (IOException e) {
            log.error("创建channel失败，" + retryInterval + "毫秒后重试。" + e.getMessage(), e);
            MqUtils.sleep(retryInterval);
            return createChannel();
        }
    }

    private void closeChannel(Channel channel) {
        try {
            channel.close();
        } catch (IOException | TimeoutException e) {
            log.warn("关闭Channel失败。", e);
        }
    }

    public void release(Channel channel) {
        if (channel == null) {
            return;
        }
        channel.clearConfirmListeners();
        channel.clearReturnListeners();
        busyChannels.remove(channel.getChannelNumber());
        boolean inserted = idleChannels.offer(channel);
        if (!inserted) {
            log.info("Channel[{}] 无法释放，将被回收。", channel.getChannelNumber());
            allChannels.remove(channel.getChannelNumber());
            closeChannel(channel);
        }
    }

    private void reset() {
        allChannels.clear();
        busyChannels.clear();
        idleChannels.clear();
        size.set(0);

        closeConnectionAsync();
    }

    private void closeConnectionAsync() {
        Thread t = new Thread(() -> connectionManager.closeConnection(namespace));
        t.start();
    }

    public int getSize() {
        return size.get();
    }
}
